package com.alison.sparkstream.sink

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

import java.sql.{Connection, DriverManager}


object E1_sink {
  """
    |我们常常需要将处理后的数据发送到各种不同的目的地，比如文件系统、数据库、消息队列等，用于展示、存储或进一步分析。本文将演示一些常见的 DStream 输出操作。
    |""".stripMargin

  def main(args: Array[String]): Unit = {
//    outputSequenceFiles
    outputRdd
  }


  def outStd() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("localhost", 9999)
    //********************************************
    // 打印结果
    inputDStream.print()
    //********************************************
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }


  """
    |saveAsTextFiles(prefix, [suffix]) 方法以 text 文件形式存储 DStream 的内容。
    |每个批次的数据都会被保存为一个单独的文件夹，文件夹名字基于参数中的 prefix 和 suffix，
    |即“prefix-Time_IN_MS[.suffix]”。
    |每个文件夹中会包含该批次的数据作为文本文件。
    |
    |文件格式
    |
    |output/streamData-时间戳.txt
    |
    |""".stripMargin

  def outputFile() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("192.168.56.104", 9999)

    // 将结果保存到文件中
    /**
     * output/streamData-时间戳.txt
     */
    inputDStream.saveAsTextFiles("file:///D:\\workspace\\lab\\learnbigdata\\learnspark\\sparkstream\\target\\classes\\" + "output/streamData", "txt")
    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  """
    |saveAsObjectFiles(prefix, [suffix])
    |
    |方法以 Java 对象序列化的方式将 Stream 中的数据保存为 SequenceFiles。
    |每个批次的数据都会被保存为一个单独的文件夹，文件夹名字基于参数中的 prefix 和 suffix，即“prefix-Time_IN_MS[.suffix]”。
    |每个文件夹中会包含该批次的数据作为文本文件。
    |
    |文件格式
    |
    |output/streamData-时间戳.seq
    |""".stripMargin

  def outputSequenceFiles() = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))
    // 从监控端口读取数据流
    val inputDStream = ssc.socketTextStream("192.168.56.104", 9999)
    // 将结果保存为 SequenceFiles
    inputDStream.saveAsObjectFiles("file:///D:\\workspace\\lab\\learnbigdata\\learnspark\\sparkstream\\target\\classes\\" + "output/streamData", "seq")
    // 启动 StreamingContext
    ssc.start()

    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

  """
    |使用通用的输出操作（foreachRDD）
    |（1）foreachRDD(func) 是最通用的输出操作，用于对 DStream 中的每个 RDD 运行任意计算。参数传入的函数 func 应该实现将每个 RDD 中的数据推送到外部系统，例如将 RDD 存入文件或通过网络写入数据库。
    |（2）下面是将一个将 Dstream 数据存到 MySQL 数据库中的样例：
    |
    |如果要将数据写入数据库，注意创建连接代码位置：
    |连接不能写在 driver 层面（序列化）
    |如果写在 foreach 则每个 RDD 中的每一条数据都创建，得不偿失；
    |增加 foreachPartition，在分区创建（获取）。
    |
    |""".stripMargin

  def outputRdd() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("outputRdd")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 创建一个 DStream，从 localhost:9999 接收输入
    val lines = ssc.socketTextStream("192.168.56.104", 9999)
    // 将每行拆分成单词
    val words = lines.flatMap(_.split(" "))
    // 统计每个单词出现的次数
    val wordCounts = words.map((_, 1)).reduceByKey(_ + _)

    // 针对每个 RDD 执行写入数据库操作
    wordCounts.foreachRDD { rdd =>
      rdd.foreachPartition { partitionOfRecords =>
        // 在每个分区内创建数据库连接
        val connection = createConnection()
        partitionOfRecords.foreach { record =>
          // 执行数据插入操作
          writeToDatabase(connection, record)
        }
        connection.close() // 关闭连接
      }
    }
    wordCounts.print()
    ssc.start()
    ssc.awaitTermination()
  }
  // 将数据写入数据库
  """
    |create database  spark default charset  utf8;
    |drop database spark;
    |
    |create database spark default character set utf8mb4 collate utf8mb4_unicode_ci;
    |
    |use spark;
    |create table word_count(word varchar(20), count int) engine =innodb charset utf8;
    |
    |""".stripMargin
  def writeToDatabase(connection: Connection, record: (String, Int)): Unit = {
    val sql = "INSERT INTO word_count (word, count) VALUES (?, ?)"
    val statement = connection.prepareStatement(sql)
    // 设置参数
    statement.setString(1, record._1)
    statement.setInt(2, record._2)
    // 执行插入操作
    statement.executeUpdate()
  }

  def createConnection() = {
    val jdbcUrl = "jdbc:mysql://localhost:3306/spark"
    val user = "root"
    val password = "root"
    // 加载 MySQL 驱动程序
    Class.forName("com.mysql.cj.jdbc.Driver")
    // 建立连接
    DriverManager.getConnection(jdbcUrl, user, password)
  }

  def outputKafka() = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("outputKafka")

    // 初始化 StreamingContext，设置微批处理的时间间隔为 5 秒
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 创建一个 DStream，从 localhost:9999 接收输入
    val lines = ssc.socketTextStream("192.168.56.104", 9999)
    // 将每行拆分成单词
    val words = lines.flatMap(_.split(" "))
    // 统计每个单词出现的次数
    val wordCounts = words.map((_, 1)).reduceByKey(_ + _)

    // Kafka 生产者配置
    val props = new java.util.Properties()
    props.put("bootstrap.servers", "192.168.60.9:9092") // Kafka 服务器地址
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")

    // 针对每个 RDD 执行发送消息到 Kafka 操作
    wordCounts.foreachRDD { rdd =>
      rdd.foreachPartition { partitionOfRecords =>
        // 在每个分区内创建Kafka生产者
        val producer = new KafkaProducer[String, String](props)

        partitionOfRecords.foreach { record =>
          // 构建消息，发送到 Kafka 主题
          val message = new ProducerRecord[String, String]("spark_topic",
            record._1, record._2.toString)
          producer.send(message)
        }

        // 关闭 Kafka 生产者
        producer.close()
      }
    }

    // 启动 StreamingContext
    ssc.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    ssc.awaitTermination()
  }

}
